package com.doitedu.core2

import com.doitedu.utils.SparkUtil
import org.apache.spark.HashPartitioner
import org.apache.spark.rdd.RDD

/**
 * @Date: 22.7.7 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C09_Checkpoint {
  def main(args: Array[String]): Unit = {
    // 设置系统 hadoop操作用户
    System.setProperty("HADOOP_USER_NAME" , "root")
    val sc = SparkUtil.getSparkContext("依赖")
    // 1 设置checkpoint的目录
    sc.setCheckpointDir("hdfs://linux01:8020/spark_ck/")

    val rdd1 = sc.textFile("data/word.txt")
    val rdd2 = rdd1.flatMap(_.split("\\s+"))
    val rdd3 = rdd2.map((_, 1))
    val rddx = rdd3.partitionBy(new HashPartitioner(rdd3.getNumPartitions))

    println(rddx.toDebugString)
   println("#######################################")
    // 只有在执行动算子的时候才会触发ck
    rddx.checkpoint()
    rddx.collect()  //  checkpoint
   // 在ck后rddx从HDFS的目录下获取数据
    val rdd4 = rddx.reduceByKey(_ + _)  //ShuffledRDD
    println(rddx.toDebugString)

    val rdd5: RDD[(String, Int)] = rdd4.map(tp => ((tp._1.toUpperCase(), tp._2)))

    val rdd6 = rdd5.groupByKey()



  }
}
